package com.study.chapter07;

import com.study.entity.Event;
import com.study.chapter05.source.ClickSource;
import org.apache.commons.lang3.StringUtils;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;

import java.sql.Timestamp;

/**
 * @Description:
 * @Author: LiuQun
 * @Date: 2022/8/4 22:30
 */
public class ProcessFuncTest {
    public static void main(String[] args) throws Exception {
        //环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

       env.addSource(new ClickSource())
               .assignTimestampsAndWatermarks(
                       WatermarkStrategy.<Event>forMonotonousTimestamps()
                       .withTimestampAssigner((data,l) -> data.timestamp)
               )
               //处理函数
               .process(new ProcessFunction<Event, String>() {
                   @Override
                   public void processElement(Event value, Context ctx, Collector<String> out) throws Exception {
                       if (StringUtils.equals(value.user,"Bob")){
                           out.collect(value.toString());
                       }else if (StringUtils.equals(value.user,"Alice")){
                           out.collect(value.user);
                       }
                       long watermark = ctx.timerService().currentWatermark();
                       System.out.println(new Timestamp(watermark));
                   }
               })
               .print();

        env.execute();
    }
}
